package com.shujia.data

import java.util.Properties

import com.google.gson.Gson
import com.shujia.bean.Car
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.io.Source

object MakeDataToKafka {
  def main(args: Array[String]): Unit = {


    //创建kafak生产者
    val properties = new Properties()

    properties.setProperty("bootstrap.servers", "node1:9092") //kafka连接地址

    properties.put("key.serializer", classOf[StringSerializer])
    properties.put("value.serializer", classOf[StringSerializer])

    val producer = new KafkaProducer[String, String](properties)


    val gson = new Gson()
    Source
      .fromFile("resources/sortData.txt")
      .getLines()
      .foreach(line => {

        //将数据格式转换成json

        val split = line.split(",")
        val carId = split(0)
        val kcId = split(1)
        val ts = split(2).toLong
        val speed = split(3).toInt
        val car = new Car(carId, kcId, ts, speed)
        val jsonStr = gson.toJson(car)
        val pr = new ProducerRecord[String, String]("ods_car", jsonStr)
        //发送数据到kafka
        producer.send(pr)

      })

    producer.close()


  }
}
